package state

import java.util
import java.util.concurrent.TimeUnit

import bean.SensorReading
import org.apache.flink.api.common.functions.{ReduceFunction, RichFlatMapFunction, RichMapFunction}
import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, MapState, MapStateDescriptor, ReducingState, ReducingStateDescriptor, ValueState, ValueStateDescriptor}
import org.apache.flink.api.common.time.Time
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.runtime.state.memory.MemoryStateBackend
import org.apache.flink.util.Collector

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/11/27 12:14
  */
object StateTest {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    //设置事件时间
    //    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    //设置状态后端
    env.setStateBackend(new MemoryStateBackend()) //内存
    env.setStateBackend(new FsStateBackend("hdfs")) //文件系统
    env.setStateBackend(new RocksDBStateBackend("mysql")) //数据库
    //设置检查点
    env.enableCheckpointing(1000L) //开启检查点
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    env.getCheckpointConfig.setCheckpointTimeout(60000L)
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(2) //设置检查点数量
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500L) //最小间隔时间
    env.getCheckpointConfig.setPreferCheckpointForRecovery(true) //优先使用检查点恢复任务
    env.getCheckpointConfig.setTolerableCheckpointFailureNumber(3) //容忍失败的次数


    //重启策略
    //时间间隔次数重启
    env.setRestartStrategy(RestartStrategies
      .fixedDelayRestart(3, 60000L))
    //
    env.setRestartStrategy(RestartStrategies
      .failureRateRestart(5,Time.of(5,TimeUnit.MINUTES),Time.of(10,TimeUnit.SECONDS)))

    //file source
    val inputPath: String = "src/main/resources/sensor.txt"
    val fileStream: DataStream[String] = env.readTextFile(inputPath)
    val socketStream = env.socketTextStream("192.168.88.106", 7777)

    //先转换为特定的类型
    val dataStream: DataStream[SensorReading] = socketStream.map(data => {
      val arr = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })
    //需求：对于温度传感器温度值跳变，超过10度，报警
    val resultStream: DataStream[(String, Double, Double)] = dataStream
      .keyBy(_.id)
      //      .flatMap(new TempChangeAlert(10.0))
      .flatMapWithState[(String, Double, Double), Double] {
      case (data: SensorReading, None) => (List.empty, Some(data.temperature))
      case (data: SensorReading, lastTemp: Some[Double]) => {
        //跟最新的温度值求差值作比较
        val diff = (data.temperature - lastTemp.get).abs
        if (diff > 10.0)
          (List((data.id, lastTemp.get, data.temperature)), Some(data.temperature))
        else
          (List.empty, Some(data.temperature))
      }
    }
    resultStream.print()

    env.execute("state test")
  }

  //实现自定义RichFlatmapFunction
  class TempChangeAlert(threshold: Double) extends RichFlatMapFunction[SensorReading, (String, Double, Double)] {
    //定义状态保存上一次的温度值
    lazy val lastTempState: ValueState[Double] = getRuntimeContext
      .getState(new ValueStateDescriptor[Double]("last-temp", classOf[Double]))

    override def flatMap(in: SensorReading, collector: Collector[(String, Double, Double)]): Unit = {
      //获取上次的温度值
      val lastTemp = lastTempState.value()
      //跟最新的温度值求差值作比较
      val diff = (in.temperature - lastTemp).abs
      if (diff > threshold)
        collector.collect((in.id, lastTemp, in.temperature))

      //更新状态
      lastTempState.update(in.temperature)

    }
  }

  //自定义ReduceFunction
  class MyReducer extends ReduceFunction[SensorReading] {
    override def reduce(t: SensorReading, t1: SensorReading): SensorReading = {
      SensorReading(t.id, t1.timestamp, t.temperature.min(t1.temperature))
    }
  }

  //Keyed state测试，必须定义RichFunction中,因为运行时上下文需要
  class MyRichMapper extends RichMapFunction[SensorReading, String] {
    var valueState: ValueState[Double] = _
    lazy val listState: ListState[Int] = getRuntimeContext
      .getListState(new ListStateDescriptor[Int]("liststate", classOf[Int]))
    lazy val mapState: MapState[String, Double] = getRuntimeContext
      .getMapState(new MapStateDescriptor[String, Double]("mapstate", classOf[String], classOf[Double]))
    lazy val reduceState: ReducingState[SensorReading] = getRuntimeContext
      .getReducingState(new ReducingStateDescriptor[SensorReading]("reducestate", new MyReducer, classOf[SensorReading]))

    override def open(parameters: Configuration): Unit = {
      valueState = getRuntimeContext
        .getState(new ValueStateDescriptor[Double]("valuestate", classOf[Double]))

    }

    override def map(in: SensorReading): String = {
      val myV = valueState.value()
      valueState.update(in.temperature)
      listState.add(1)
      val list = new util.ArrayList[Int]()
      list.add(2)
      list.add(3)
      listState.addAll(list)
      listState.update(list) //替换
      mapState.contains("sensor_1")
      mapState.get("sensor_1")
      mapState.put(in.id, in.temperature)
      reduceState.get()
      reduceState.add(in)

      in.id
    }
  }

}
